import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class Sender implements Runnable {

	private static final int MAX_SIZE_MSG = 10000;
	private User recipient;
	private Message msg;
	private BlockingQueue<Message> queue;
	private long senderID;
	
	public Sender(Message msg, User recipient, ArrayBlockingQueue<Message> queue, long senderID) {
		this.msg = msg;
		this.recipient = recipient;
		this.queue = queue;
		this.senderID = senderID;
		this.msg.setSenderID(senderID);
		new Thread(this, "Sender "+senderID).start();
	}
	
	public Sender(Message msg, User recipient, ArrayBlockingQueue<Message> queue) {
		this.msg = msg;
		this.recipient = recipient;
		this.queue = queue;
		new Thread(this, "Sender").start();
	}

	public static byte[] getBytesFromObj(Object obj) throws IOException {
		ByteArrayOutputStream baos = new ByteArrayOutputStream();
		ObjectOutputStream oos = new ObjectOutputStream(baos);
		oos.writeObject(obj);
		oos.flush();
		// get the byte array of the object
		byte[] buf = baos.toByteArray();
		return buf;
	}

	@Override
	public void run() {
		if (msg instanceof NormalMessage)
			try {
				rUnicast((NormalMessage) msg);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		else {
			System.out.println("Sending AckMessage with ID "+msg.getSenderID());
			unicast((AckMessage)msg);
		}
	}

	public void rUnicast(NormalMessage msg) throws InterruptedException {
		System.out.println("Sending NormalMessage with ID "+msg.getSenderID());
		unicast(msg);
		long sTimer = System.currentTimeMillis();
		long eTimer = sTimer + Configuration.getInstance().getMsgTimeOut();
		while (true) {
//			if (!queue.isEmpty()) {
				AckMessage aMsg = (AckMessage) queue.peek();
				if (aMsg != null && senderID == aMsg.getSenderID()) {
					System.out.println("MessageID " + aMsg.getSenderID() + " taking");
					System.err.println("Receive Ack");
					queue.take();
					break;
				}
//			}
			if (System.currentTimeMillis() >= eTimer) {
				System.out.println("Resending NormalMessage with ID "+msg.getSenderID());
				unicast(msg);
				sTimer = System.currentTimeMillis();
				eTimer = sTimer + Configuration.getInstance().getMsgTimeOut();
				//System.out.println("timer " + sTimer + " " + eTimer);
			}
			Thread.sleep(100); //sleep for a while before checking the queue again
		}
	}
	
	public void unicast(Message msg) {
		Random rand = new Random();
		try (DatagramSocket clientSocket = new DatagramSocket()) {
			int num = rand.nextInt() % 100;
			num = num < 0 ? -num : num;
			//System.out.println("--------" + num);
			if (num < 100 * Configuration.getInstance().getMsgDropRate()) {
				return;
			} else {
				Thread.sleep(num * (Configuration.getInstance().getMsgMaxDelay()/100));
			}
			InetAddress IPAddress = InetAddress.getByName(recipient.getIp());
			byte[] sendData = new byte[MAX_SIZE_MSG];
//			System.out.println("Sender: " + msg.getSenderID());
			sendData = getBytesFromObj(msg);
			DatagramPacket sendPacket = new DatagramPacket(sendData,
					sendData.length, IPAddress, recipient.getPort());
			clientSocket.send(sendPacket);
			//System.out.println("Sent msg from " + msg.getSender() + " to " + recipient);
		} catch (Exception ex) {
			ex.printStackTrace();
		}
	}
}